package org.unigram.oluolu.recommend;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;

import org.unigram.oluolu.recommend.function.IHashFunction;

/**
 *
 */
public class SelectClustersMapper extends
    Mapper<LongWritable, Text, LongWritable, Text> {

    /**
     * map method.
     * 
     * @param key dummy
     * @param value containing query information (user id, query string, time)
     * @param context write keys and values which are input of reducer
     * @throws IOException -
     * @throws InterruptedException -
     */
    @Override
    public final void map(final LongWritable key,
            final Text value, final Context context)
            throws IOException, InterruptedException {

        String inputStr = value.toString();

        try {
            String[] tokens = inputStr.split("\t");
             Map<String, Long> featureMap 
                = this.extractFeatures(tokens[1]);
            context.write(this.function.returnClusterId(featureMap), 
                    new Text(tokens[0]));
        } catch (ArrayIndexOutOfBoundsException e) {
            System.out.println("PARSING ERROR in line: " + inputStr);
            e.printStackTrace();
        }
        
    }

    /** 
     * Extract set of features.
     *  
     * @param featureStr string containing set of features
     * @return Map (key: feature, value: count)  
     */
    private Map<String, Long> extractFeatures(
            final String featureStr) {
        Map<String, Long> rtMap = new HashMap<String, Long>();
        String[] featureArray = featureStr.split(":");
        for (int i=0; i<featureArray.length; i++) {
            rtMap.put(featureArray[0], new Long(1));
        }
        return rtMap;
    }
    
    
    /**
     * setup.
     * 
     * @param context contains configuraiton
     */
    public final void setup(final Context context) {
        Configuration jc = context.getConfiguration();
        /** create a object implements IHashFunction */
        String functionClassName = RecommendConstants.DEFAULT_HASH_FUNCTION;
        if (context == null || jc == null) {
            /* added default configuration for testing */
            jc = new Configuration();
        }
        try {
            functionClassName = jc.get(RecommendConstants.HASH_FUNCTION,
                    RecommendConstants.DEFAULT_HASH_FUNCTION);
            Class<? extends IHashFunction> functionClass = Class.forName(
                    functionClassName).asSubclass(IHashFunction.class);
            Constructor<? extends IHashFunction> constructor = functionClass
                    .getConstructor(Configuration.class);
            function = constructor.newInstance(jc);
        } catch (NoSuchMethodException nsme) {
            throw new RuntimeException(nsme);
        } catch (ClassNotFoundException cnfe) {
            throw new RuntimeException(cnfe);
        } catch (InstantiationException ie) {
            throw new RuntimeException(ie);
        } catch (IllegalAccessException iae) {
            throw new RuntimeException(iae);
        } catch (InvocationTargetException ite) {
            throw new RuntimeException(ite.getCause());
        }        
    }    


    /** Hash function object. */
    private IHashFunction function;    
    
}
